package jiang.jredis.channel;

import lombok.SneakyThrows;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.function.Consumer;

/**
 * 动态代理类
 * 代理Channel接口的方法
 * @see Channel
 */
public class ChannelInvocationHandler implements InvocationHandler {

    private final String topic;
    private final RedisTemplate<String, Object> redisTemplate;
    private final RedisMessageListenerContainer redisMessageListenerContainer;

    /**
     * 消费者和线程池列表
     */
    private final Map<Consumer<Object>,Executor> consumerExecutorMap;

    /**
     * 阻塞获取消息的线程会添加future到队列中
     */
    private final BlockingQueue<CompletableFuture<Object>> futureBlockingQueue;

    /**
     * Redis消息监听器
     * 1、让已添加的消费者处理消息
     * 2、给阻塞的接收线程返回数据
     */
    private final MessageListener messageListener;

    /**
     * 构造方法
     * 1.注入属性
     * 2.初始化成员变量
     * 3.注册监听器
     */
    public ChannelInvocationHandler(String topic, RedisTemplate<String, Object> redisTemplate, RedisMessageListenerContainer redisMessageListenerContainer) {
        // 注入属性
        this.topic = topic;
        this.redisTemplate = redisTemplate;
        this.redisMessageListenerContainer = redisMessageListenerContainer;
        // 初始化成员变量
        this.consumerExecutorMap = new ConcurrentHashMap<>();
        this.futureBlockingQueue = new LinkedBlockingQueue<>();
        this.messageListener = initMessageListener();
        // 注册监听器
        redisMessageListenerContainer.addMessageListener(messageListener, ChannelTopic.of(topic));
    }

    /**
     * 动态代理方法入口
     */
    @SneakyThrows
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws InvocationTargetException, IllegalAccessException, InterruptedException {

        /**
         * 如果不是Channel接口，则不代理
         */
        if(proxy instanceof Channel == false) {
            return method.invoke(args);
        }

        /**
         * 代理publish方法
         */
        if ("publish".equals(method.getName()) && args.length == 1) {

            // 获取消息对象
            Object message = args[0];

            // 发布
            redisTemplate.convertAndSend(topic, message);

            // 返回
            return null;
        }

        /**
         * 代理publishAsync方法
         */
        else if ("publishAsync".equals(method.getName()) && args.length == 2) {

            // 获取消息对象和线程池
            Object message = args[0];
            Executor executor = (Executor) args[1];

            // 使用线程池异步发布
            executor.execute(() -> {
                redisTemplate.convertAndSend(topic, message);
            });

            // 返回
            return null;
        }

        /**
         * 代理subscribe方法
         */
        else if("subscribe".equals(method.getName()) && args.length == 2) {

            // 获取消费者和线程池
            Consumer<Object> consumer = (Consumer) args[0];
            Executor executor = (Executor) args[1];

            // 将消费者和线程池添加到消费者线程池map中
            consumerExecutorMap.put(consumer, executor);

            // 返回
            return null;
        }

        /**
         * 代理unSubscribe方法
         */
        else if ("unSubscribe".equals(method.getName()) && args.length == 1) {

            // 获取消费者对象
            Consumer<Object> consumer = (Consumer) args[0];

            // 将消费者从消费者线程池map中移除
            consumerExecutorMap.remove(consumer);

            // 返回
            return null;
        }

        /**
         * 代理receive方法 永久等待
         */
        else if("receive".equals(method.getName()) && (args == null || args.length == 0)) {

            // 创建CompletableFuture用于异步获取数据
            CompletableFuture<Object> future = new CompletableFuture<>();

            // 将future对象放入队列中等待异步被取出执行
            futureBlockingQueue.put(future);

            // 阻塞直到有数据返回
            return future.join();
        }

        /**
         * 代理receive方法 等待给定时间
         */
        else if ("receive".equals(method.getName()) && args.length == 2) {

            // 获取等待时间变量
            long timeout = (long) args[0];
            TimeUnit unit = (TimeUnit) args[1];

            // 创建CompletableFuture用于异步获取数据
            CompletableFuture<Object> future = new CompletableFuture<>();

            // 将future对象放入队列中等待异步被取出执行
            futureBlockingQueue.put(future);

            // 阻塞给定时间获取数据
            return future.get(timeout, unit);
        }

        /**
         * 代理close方法
         */
        else if ("close".equals(method.getName())  && (args == null || args.length == 0)) {

            // 移除监听器
            redisMessageListenerContainer.removeMessageListener(messageListener);

            // 返回
            return null;
        }

        return null;
    }

    /**
     * Redis消息监听器
     * 1、让已添加的消费者处理消息
     * 2、给阻塞的接收线程返回数据
     */
    private MessageListener initMessageListener() {
        return (message, pattern) -> {

            // 获取消息
            byte[] bytes = message.getBody();

            // 1、让已添加的消费者处理消息
            for (Map.Entry<Consumer<Object>,Executor> entry : consumerExecutorMap.entrySet()) {

                // 获取消费者对象和执行任务的线程池
                Consumer<Object> consumer = entry.getKey();
                Executor executor = entry.getValue();

                // 反序列化消息
                Object data = redisTemplate.getValueSerializer().deserialize(bytes);

                // 交由线程池执行任务
                executor.execute(() -> {
                    consumer.accept(data);
                });
            }

            // 2、为每一个future都创建一份数据返回
            int n = futureBlockingQueue.size();
            List<CompletableFuture<Object>> futureList = new ArrayList<>(n);

            // 取出队列元素并遍历
            futureBlockingQueue.drainTo(futureList, n);
            for (CompletableFuture<Object> future : futureList) {

                // 反序列化消息
                Object data = redisTemplate.getValueSerializer().deserialize(bytes);

                // 将对象通过future返回
                future.complete(data);
            }
        };
    }

}
